package com.shujia.stream

import com.shujia.util.FlinkTableEnv
import org.apache.flink.table.api.TableEnvironment

object JIChaControl {
  def main(args: Array[String]): Unit = {
    val table: TableEnvironment = FlinkTableEnv.getTableEnv(5000)


    /**
      * 卡口过车数据
      */
    table.executeSql(
      """
        |CREATE TABLE cars_kafka (
        |  `car` STRING,
        |  `city_code` STRING,
        |  `county_code` STRING,
        |  `card` BIGINT,
        |  `camera_id` STRING,
        |  `orientation` STRING,
        |  `road_id` BIGINT,
        |  `time` BIGINT,
        |  `speed`  DOUBLE,
        |  `ts` AS PROCTIME()
        |) WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'cars',
        |  'properties.bootstrap.servers' = 'master:9092',
        |  'properties.group.id' = 'JIChaControl',
        |  'scan.startup.mode' = 'earliest-offset',
        |  'format' = 'json',
        |  'json.ignore-parse-errors' = 'true'
        |)
        |
      """.stripMargin)

    /**
      * 创建mysql  source表获取布控列表
      *
      */

    table.executeSql(
      """
        |CREATE TABLE control_list (
        |  car STRING,
        |  PRIMARY KEY (car) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/car?useUnicode=true&characterEncoding=UTF-8',
        |   'table-name' = 'control_list',
        |   'username' = 'root',
        |   'password' = '123456'
        |)
        |
      """.stripMargin)

    /**
      * 创建sink表
      *
      */

    table.executeSql(
      """
        |
        |CREATE TABLE control_cars (
        |  `car` STRING,
        |  `city_code` STRING,
        |  `county_code` STRING,
        |  `card` BIGINT,
        |  `camera_id` STRING,
        |  `orientation` STRING,
        |  `road_id` BIGINT,
        |  `time` BIGINT,
        |  `speed`  DOUBLE,
        |  PRIMARY KEY (car,`time`) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/car?useUnicode=true&characterEncoding=UTF-8',
        |   'table-name' = 'control_cars',
        |   'username' = 'root',
        |   'password' = '123456'
        |)
        |
        |
      """.stripMargin)


    table.executeSql(
      """
        |
        |insert into control_cars
        |select
        |a.car as car,
        |city_code,
        |county_code,
        |card,
        |camera_id,
        |orientation,
        |road_id,
        |`time`,
        |speed
        |from
        |cars_kafka as a
        |inner join
        |control_list  FOR SYSTEM_TIME AS OF a.ts as b
        |on a.car=b.car
        |
        |
      """.stripMargin)

  }

}
